package cn.tpshion.pulsar.listener.support;

import cn.tpshion.pulsar.annoation.PulsarConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.shade.org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class PulsarConsumerHandle implements ApplicationContextAware, CommandLineRunner {

    private ApplicationContext context;

    private static final Map<Method, Object> handleBeanMap = new ConcurrentHashMap<>();

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;
    }

    public void handle(Consumer consumer, Message message){
        String topic = consumer.getTopic();
        for (Map.Entry<Method, Object> entry : handleBeanMap.entrySet()) {
            Method method = entry.getKey();
            Object bean = entry.getValue();
            PulsarConsumer pulsarConsumer = method.getAnnotation(PulsarConsumer.class);
            if(!topic.endsWith(pulsarConsumer.topic())){
                continue;
            }
            String consumerName = consumer.getConsumerName();
            if(!StringUtils.equals(consumerName, pulsarConsumer.consumerName())){
                continue;
            }
            if(pulsarConsumer.autoAck()){
                autoHandle(consumer, message, method, bean);
            }else{
                ReflectionUtils.invokeMethod(method, bean, consumer, message);
            }
        }
    }

    private void autoHandle(Consumer consumer, Message message, Method method, Object bean){
        MessageId messageId = message.getMessageId();
        try {
            byte[] data = message.getData();
            log.info("autoHandle exec messageId:{} data:{}", messageId, new String(data));
            ReflectionUtils.invokeMethod(method, bean, consumer, message);
            consumer.acknowledge(message);
        } catch (PulsarClientException e) {
            consumer.negativeAcknowledge(message);
            log.error("autoHandle messageId:{} error:{}", messageId, e.getMessage(), e);
        }
    }

    @Override
    public void run(String... args) throws Exception {
        String[] beanDefinitionNames = context.getBeanDefinitionNames();
        for (String beanDefinitionName : beanDefinitionNames) {
            Object bean = context.getBean(beanDefinitionName);
            Method[] methods = bean.getClass().getDeclaredMethods();
            for (Method method : methods) {
                if(method.isAnnotationPresent(PulsarConsumer.class)){
                    handleBeanMap.put(method, bean);
                }
            }
        }
    }
}
